package com.remoter.provider.internal;

import java.net.BindException;
import java.net.InetSocketAddress;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.remoter.api.configure.IConfiguration;
import com.remoter.api.configure.support.AbstractConfiguration;
import com.remoter.api.extension.annotation.ExtensionName;
import com.remoter.api.extension.support.ExtensionLoader;
import com.remoter.api.provider.IProviderService;
import com.remoter.api.transport.IMessageTask;
import com.remoter.api.transport.IMessageTaskCallBack;
import com.remoter.api.transport.IServer;
import com.remoter.api.util.AbortPolicyWithReport;
import com.remoter.api.util.Final;
import com.remoter.api.util.NamedThreadFactory;
import com.remoter.provider.internal.util.FinalProviderInternal;

@ExtensionName("internal")
public class InternalProviderService implements IProviderService{
	
	private static final Logger logger = LoggerFactory.getLogger(InternalProviderService.class);
	private static volatile ListeningExecutorService listeningExecutorService;
	
	private IConfiguration configuration;
	private InetSocketAddress bind;
	private IServer server;
	
	public InternalProviderService(){
		this.configuration = AbstractConfiguration.getConfiguration();
	}
	
	@Override
	public void bind(InetSocketAddress bind) throws BindException, InterruptedException{
		if(null != this.server){
			if(this.server.isAvailable()){
				logger.debug("server is started");
				return;
			}
		}
		this.bind = bind;
		this.server = ExtensionLoader.getService(IServer.class,this.configuration.getOption(Final.O_SERVER_TRANSPORT));
		this.server.bind(this.bind);
		logger.info("server bind success : " + this.bind);
	}
	
	@Override
	public void unBind(){
		if(null != this.server){
			this.server.disConnect();
		}
	}

	@Override
	public void submit(IMessageTask messageTask, IMessageTaskCallBack messageTaskCallBack) {
		if(null == listeningExecutorService){
			synchronized(InternalProviderService.class){
				if(null == listeningExecutorService){
					String taskName = this.configuration.getOption(FinalProviderInternal.O_PROVIDER_INTERNAL_TASK_NAME);
					ThreadPoolExecutor executor = new ThreadPoolExecutor(
							Final.PROCRESS,Final.PROCRESS*100,
							1000,
							TimeUnit.MILLISECONDS,
			                new LinkedBlockingQueue<Runnable>(Final.PROCRESS*500),
			                new NamedThreadFactory(taskName,true),new AbortPolicyWithReport(taskName));
					listeningExecutorService = MoreExecutors.listeningDecorator(executor);
				}
			}
		}
		ListenableFuture<Boolean> future = listeningExecutorService.submit(messageTask);
		Futures.addCallback(future,messageTaskCallBack,listeningExecutorService);
	}

	@Override
	public InetSocketAddress getBindSocketAddress() {
		return new InetSocketAddress(this.configuration.getOption(FinalProviderInternal.O_PROVIDER_INTERNAL_HOST),this.configuration.getOption(FinalProviderInternal.O_PROVIDER_INTERNAL_PORT));
	}
	
}